package com.demo01_video.consumer;


import com.demo01_video.bean.Video;
import com.demo01_video.client.FeedServiceClient;
import com.demo01_video.repository.VideoRepository;
import com.demo01_video.uitl.FfmpegUtil;
import com.demo01_video.uitl.MinioUtil;
import lombok.RequiredArgsConstructor;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.io.File;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

// 消费者
@Component
@RequiredArgsConstructor
public class VideoConsumer {
    private final VideoRepository videoRepository;
    private final FeedServiceClient feedServiceClient;
    private final FfmpegUtil ffmpegUtil;
    private final MinioUtil minIOUtil;
    private final KafkaTemplate<String, Object> kafkaTemplate;

    @KafkaListener(topics = "video-publish")
    public void consumeVideoPublish(String videoId) {
        Long videoIdLong = Long.valueOf(videoId);
        Video video = videoRepository.findById(videoIdLong).orElseThrow();
        
        try {
            // 1. 更新状态为「转码中」
            video.setStatus(1); // 1表示转码中
            videoRepository.save(video);
            
            // 2. 从MinIO下载原始视频
            String originalVideoKey = extractKeyFromUrl(video.getVideoUrl()); // 从URL中提取MinIO对象键
            File originalVideoFile = minIOUtil.downloadFile(originalVideoKey);
            
            // 3. 准备转码后的输出文件
            File transcodedVideoFile = File.createTempFile("transcoded-", ".mp4");
            
            // 4. 执行转码
            boolean success = ffmpegUtil.convertVideo(originalVideoFile, transcodedVideoFile);
            
            if (success) {
                // 5. 上传转码后的视频到MinIO
                String transcodedVideoKey = "transcoded/" + UUID.randomUUID() + ".mp4";
                // 使用MinioUtil的uploadFile方法上传转码后的视频
                String transcodedVideoUrl = minIOUtil.uploadFile(transcodedVideoFile, transcodedVideoKey);
                
                // 6. 更新视频信息
                video.setVideoUrl(transcodedVideoUrl); // 更新为转码后的视频地址
                video.setStatus(2); // 2表示转码成功
                videoRepository.save(video);
                
                // 7. 发送消息到feed-push主题，异步处理推送
                Map<String, Object> feedPushMessage = new HashMap<>();
                feedPushMessage.put("videoId", video.getId());
                feedPushMessage.put("authorId", video.getAuthorId());
                feedPushMessage.put("timestamp", System.currentTimeMillis());
                kafkaTemplate.send("feed-push", String.valueOf(video.getAuthorId()), feedPushMessage);
            } else {
                // 转码失败
                video.setStatus(3); // 3表示转码失败
                videoRepository.save(video);
            }
            
            // 8. 清理临时文件
            originalVideoFile.delete();
            transcodedVideoFile.delete();
            
        } catch (Exception e) {
            // 处理任何异常
            video.setStatus(3); // 3表示转码失败
            videoRepository.save(video);
            e.printStackTrace();
        }
    }
    
    private String extractKeyFromUrl(String url) {
        // 实现从MinIO URL中提取对象键的逻辑
        // 例如: http://localhost:9000/bucket-name/object-key -> object-key
        return url.substring(url.lastIndexOf("/") + 1);
    }
}